package drds.binlog.parse.index;

import drds.binlog.common.position.ClientInfo;
import drds.binlog.common.position.LogPosition;
import drds.binlog.metadata.MetaDataManager;
import drds.binlog.parse.exception.BinlogParseException;
import drds.binlog.store.LogPositions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.List;


public class MetaLogPositionManager extends AbstractLogPositionManager
{

    private final static Logger logger = LoggerFactory.getLogger(MetaLogPositionManager.class);

    private final MetaDataManager metaDataManager;

    public MetaLogPositionManager(MetaDataManager metaDataManager)
    {
        if (metaDataManager == null)
        {
            throw new NullPointerException("null metaDataManager");
        }

        this.metaDataManager = metaDataManager;
    }

    @Override
    public void stop()
    {
        super.stop();

        if (metaDataManager.isRunning())
        {
            metaDataManager.stop();
        }
    }

    @Override
    public void start()
    {
        super.start();

        if (!metaDataManager.isRunning())
        {
            metaDataManager.start();
        }
    }

    @Override
    public LogPosition getLatestIndexBy(String destination)
    {
        List<ClientInfo> clientIdentities = metaDataManager.getClientInfoList(destination);
        LogPosition result = null;
        if (!CollectionUtils.isEmpty(clientIdentities))
        {
            // 尝试找到一个最小的logPosition
            for (ClientInfo clientInfo : clientIdentities)
            {
                LogPosition position = (LogPosition) metaDataManager.getPosition(clientInfo);
                if (position == null)
                {
                    continue;
                }

                if (result == null)
                {
                    result = position;
                } else
                {
                    result = LogPositions.min(result, position);
                }
            }
        }

        return result;
    }

    @Override
    public void persistLogPosition(String destination, LogPosition logPosition) throws BinlogParseException
    {
        // do nothing
        logger.info("persist LogPosition:{}", destination, logPosition);
    }
}
